package com.myelasticsearch.zxisl;

import com.google.common.collect.Maps;
import com.myelasticsearch.utils.ESUtils;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

import java.io.IOException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;


public class InsertT {

    public static void execute(TransportClient client, String index, String type, List<HashMap<String, Object>> datas) throws IOException, ParseException, InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();
        System.out.println("execute start: " + startTime);
//        if(!index_exists(index)){
//			create_index_type_mapping(index,type);
//			bulk_post_document(index,type,datas);
//		}else{
        bulk_post_document(client, index, type, datas);
//        }
        long endTime = System.currentTimeMillis();
        System.out.println("execute end: " + (endTime - startTime));
    }

    /**
     * 判断索引是否存在
     */
    private static boolean index_exists(String index) {
        TransportClient client = ESUtils.getTransportClient();
        ActionFuture<IndicesExistsResponse> response = client.admin().indices().exists(Requests.indicesExistsRequest(index));
        return response.actionGet().isExists();
    }

    /**
     * 创建索引
     */
    public static void create_index_type_mapping(TransportClient client, String index, String type) throws IOException {
        System.out.println("create_index_type_setting_mapping......");
        XContentBuilder builder_mapping = XContentFactory.jsonBuilder().startObject()
                .startObject(type)      //type名
                .startObject("properties")        //属性
                .startObject("id").field("type", "string").endObject()       //字段
                .startObject("platform").field("type", "long").endObject()
                .startObject("username").field("type", "string").endObject()
                .startObject("nickname ").field("type", "string").endObject()
                .startObject("url").field("type", "string").endObject()
                .startObject("isharmful").field("type", "boolean").endObject()
                .startObject("content").field("type", "string").field("analyzer","ik").field("search_analyzer","ik").endObject()
                .startObject("title").field("type", "string").field("analyzer","ik").field("search_analyzer","ik").endObject()
//                .startObject("title").field("type","string").field("analyzer","ik").field("search_analyzer","ik").endObject()
                .startObject("pic_url").field("type", "string").endObject()
                .startObject("timestamp").field("type", "string").endObject()
                .startObject("source_id").field("type", "long").endObject()
                .startObject("lasttime").field("type", "string").endObject()
                .startObject("identify_id").field("type", "long").endObject()
                .startObject(" first_time").field("type", "string").endObject()
                .startObject("update_time").field("type", "string").endObject()
                .startObject("ip").field("type", "string").endObject()
                .startObject("location").field("type", "string").endObject()
                .startObject("source_name").field("type", "string").endObject()
                .startObject("source_type").field("type", "long").endObject()
                .startObject("country_code").field("type", "long").endObject()
                .startObject("location_code").field("type", "long").endObject()
                .startObject("province_code").field("type", "long").endObject()
                .startObject("_version_").field("type", "long").endObject()
                .endObject()
                .endObject().endObject();

        Map<String, String> settings = Maps.newHashMap();
        settings.put("number_of_shards", "5");
        settings.put("number_of_replicas", "1");

        ActionFuture<CreateIndexResponse> response = client.admin().indices().create(Requests.createIndexRequest(index).settings(settings).mapping(type, builder_mapping));
        System.out.println("isAcknowledged: " + response.actionGet().isAcknowledged());
    }


    /**
     * 批量添加索引数据
     *
     * @throws ParseException
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static void bulk_post_document(TransportClient client, String index, String type, List<HashMap<String, Object>> datas) throws ParseException, InterruptedException, ExecutionException {
        BulkRequest request = Requests.bulkRequest();
        for (Map<String, Object> data : datas) {
            request.add(Requests.indexRequest(index).type(type).source(data));
        }
        //执行批量处理
        client.bulk(request);
    }
}